/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.tabletserver;

import java.io.ByteArrayOutputStream;
import java.io.DataOutput;
import java.io.DataOutputStream;
import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.SortedSet;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.nhncorp.neptune.changelog.ChangeLogInterruptedException;
import com.nhncorp.neptune.client.Cell;
import com.nhncorp.neptune.client.CellFilter;
import com.nhncorp.neptune.client.NTable;
import com.nhncorp.neptune.client.Row;
import com.nhncorp.neptune.client.RowFilter;
import com.nhncorp.neptune.common.Constants;
import com.nhncorp.neptune.common.IdGenerator;
import com.nhncorp.neptune.common.NeptuneLock;
import com.nhncorp.neptune.common.aop.ProxyFactory;
import com.nhncorp.neptune.common.conf.NConfiguration;
import com.nhncorp.neptune.common.exception.ChangeLogException;
import com.nhncorp.neptune.common.exception.TxException;
import com.nhncorp.neptune.common.ipc.NRPC;
import com.nhncorp.neptune.common.util.NetworkUtil;
import com.nhncorp.neptune.fs.ChangeLogFileSystemIF;
import com.nhncorp.neptune.fs.GPath;
import com.nhncorp.neptune.fs.NeptuneFileSystem;
import com.nhncorp.neptune.tablet.ColumnValue;
import com.nhncorp.neptune.tablet.RowColumnValues;
import com.nhncorp.neptune.tablet.TableSchema;
import com.nhncorp.neptune.tablet.TabletInfo;
import com.nhncorp.neptune.tablet.TabletReport;
import com.nhncorp.neptune.tabletserver.action.ActionChecker;
import com.nhncorp.neptune.tabletserver.action.ApplyAction;
import com.nhncorp.neptune.tabletserver.action.CommitAction;
import com.nhncorp.neptune.tabletserver.action.MajorCompactionAction;
import com.nhncorp.neptune.tabletserver.action.MinorCompactionAction;
import com.nhncorp.neptune.tabletserver.action.MinorCompactionActionIF;
import com.nhncorp.neptune.tabletserver.action.ScannerAction;
import com.nhncorp.neptune.tabletserver.action.TabletSplitAction;
import com.nhncorp.neptune.tabletserver.action.TxAction;
import com.nhncorp.pleiades.client.LockService;
import com.nhncorp.pleiades.protocol.LockMetaData;

/**
 * commit: ChangeLogMerge를 하면 안됨
 * scanner opened: Major Compaction, Minor Compaction, Split or Merge 못함. drop, stop은 기다렸다가 close되면 시작
 * MinorCompaction: MajorCompaction, Split, Merge, ScanOpen 못하고, drop, stop은 기다렸다가 종료되면 시작
 * MajorCompaction: MinorCompaction, Split, Merge, ScanOpen 못하고, drop, stop은 기다렸다가 종료되면 시작
 * Split or Merge: Major Compaction, Minor Compaction, drop, stop, ScanOpen 못함. commit은 Split/Merge 작업중 MemTablet에 관련된 작업 진행중에만 못함
 * drop or stop: all stop
 * @author babokim
 *
 */
public class Tablet implements Constants {
  public static final String DIR_TEMP_SPLIT = "/temp/split/";
  public static final String DIR_TEMP_MINOR = "/temp/minor_compaction/";
  public static final String DIR_TEMP_MAJOR = "/temp/major_compaction/";
  public static final String DIR_TEMP_UPLOAD = "/temp/upload/";
  
  //FIXME DiskSSTable에 대한 Cache 기능: 사용자의 정의에 따라 특정 Table의 경우 DiskSSTable의 모든 데이터를 메모리에 올려 서비스 한다.
  public static final Log LOG = LogFactory.getLog(Tablet.class.getName());
  //public static Random rand = new Random();
  
  protected final TabletInfo tabletInfo;

  protected final MemorySSTableIF memorySSTable;
  
  protected DiskSSTableIF diskSSTable;
  
  protected final Map<TxId, TxValue> txMap = Collections.synchronizedMap(new HashMap<TxId, TxValue>());

  //lockedRow에 대해 주기적으로 돌면서 아무런 반응이 없는 상태에서 lock을 오랫동안 잡고 있는 경우 해제 -> TabletServer의 TxListener에서 처리
  //rowkey, lock time
  protected final Map<Row.Key, Long> lockedRow = Collections.synchronizedMap(new HashMap<Row.Key, Long>()); 
  
  protected NConfiguration conf;
  
  //private RecordSearcher recordSearcher;
  
  //private Integer changeLogFileCount = 0;
  
  private NeptuneFileSystem fs;
  
  // scannerId, TabletScanner
  private final Map<String, TabletScanner> tabletScanners = Collections.synchronizedMap(new HashMap<String, TabletScanner>());

  private boolean ready = false;

  protected final ActionChecker actionChecker;
  
  protected final NeptuneLock lock = new NeptuneLock();
  //private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
  
  protected final TableSchema table;
  
  protected long maxTabletSizeByte;
  
  protected MinorCompactionActionIF minorCompactionAction;
  
  protected MajorCompactionAction majorCompactionAction;
  
  protected TabletSplitAction tabletSplitAction; 
  
  //protected TabletDropActionIF tabletDropAction;
  
  protected boolean testMode = false;
  
  protected TabletServer tabletServer;
  
  private Row.Key endRowKey;
  
  private boolean droped;
  
  private AtomicBoolean inChangeLogError = new AtomicBoolean(false);
  
  public Tablet(
      NConfiguration conf, 
      TableSchema table, 
      TabletServer tabletServer, 
      TabletInfo tabletInfo) throws IOException {
    
    this.tabletServer = tabletServer;
    this.conf = conf;
    this.testMode = conf.getBoolean("testmode", false);
    this.fs = NeptuneFileSystem.get(conf);
    
    if(table == null) {
      throw new IOException("No table info:" + tabletInfo);
    }
    if(tabletServer != null) {
      this.actionChecker = new ActionChecker(
          tabletServer.compactionExecutor, 
          tabletServer.splitExecutor,
          tabletServer.actionExecutor,
          tabletServer.threadGroup,
          tabletInfo); 
          //tabletServer.tabletActionListener);
    } else {
      this.actionChecker = new ActionChecker(null, null, null, null, tabletInfo);
    }
    this.tabletInfo = tabletInfo;
    
    this.endRowKey = tabletInfo.getEndRowKey();
    
    if(testMode) {
      memorySSTable = (MemorySSTableIF)ProxyFactory.createProxyClass("memorySSTable");
    } else {
      memorySSTable = new MemorySSTable();
    }
    memorySSTable.init(tabletServer, conf, tabletInfo, table.getNumOfVersion());
    
    if(testMode) {
      diskSSTable = (DiskSSTableIF)ProxyFactory.createProxyClass("diskSSTable");
    } else {
      diskSSTable = new DiskSSTable();
    }
    diskSSTable.init(tabletServer, diskSSTable, conf, tabletInfo, table.getNumOfVersion());
    
//    if(testMode) {
//      tabletDropAction = (TabletDropActionIF)ProxyFactory.createProxyClass("tabletDropAction");
//    } else {
//      tabletDropAction = new TabletDropAction();
//    }    
//    if(tabletServer != null) {
//      tabletDropAction.init(tabletServer, tabletDropAction, 
//                              LockServiceFactory.getLockService(conf, tabletServer.getHostName(), true), tabletInfo);
//    }
    
    this.table = table;
    
    float maxTabletSize = conf.getFloat("maxTabletSize", 120.0f);
    maxTabletSizeByte = (long)(maxTabletSize * 1024.0f * 1024.0f);
    
    if(testMode) {
      this.minorCompactionAction = (MinorCompactionActionIF)ProxyFactory.createProxyClass("minorCompactionAction");
    } else {
      this.minorCompactionAction = new MinorCompactionAction();
    }
    this.minorCompactionAction.init(minorCompactionAction, conf, this);

    this.majorCompactionAction = new MajorCompactionAction(this);
    this.tabletSplitAction = new TabletSplitAction(this, tabletServer);
  }
  
  public TableSchema getTable() {
    return table;
  }
  
  public NConfiguration getConf() {
    return conf;
  }
  
  public void setDiskSSTable(DiskSSTableIF diskSSTable) {
    lock.obtainWriteLock();
    try {
      this.diskSSTable = diskSSTable;
      //this.recordSearcher = new RecordSearcher(new Searchable[]{memorySSTable, this.diskSSTable});
    } finally {
      lock.releaseWriteLock();
    }
  }
  
  public void print() throws IOException {
    System.out.println(tabletInfo.toString());
    System.out.println("------------------ Memory Data -----------------");
    memorySSTable.print();
//    System.out.println("------------------ Disk Data -----------------");
//    diskSSTable.print();
  }

  /**
   * 
   * @return
   * @throws IOException
   */
  public Row.Key[] getLockedRowKey() throws IOException {
    lock.obtainReadLock();
    try {
      return lockedRow.keySet().toArray(new Row.Key[]{});
    } finally {
      lock.releaseReadLock();
    }
  }
  
  /**
   * Tablet이 TabletServer에 할당되면 MapFile의 index 파일을 읽어 index를 구성하고 
   * ChangeLog 파일을 읽어 MemoryTablet을 구성한다.
   * @throws IOException
   */
  public boolean load() throws IOException {
    //long startTime = System.currentTimeMillis();

    //load change log file and add MemoryTablet
    boolean memoryLoadResult = memorySSTable.loadFromChangeLog();
    
    //load disk tablet index
    diskSSTable.load();
    
    return memoryLoadResult;
  }

  /**
   * 데이터를 조회한다.
   * @param rowKey
   * @param columnName
   * @param columnKey
   * @return
   * @throws IOException
   */
  public ColumnValue get(Row.Key rowKey, String columnName, Cell.Key cellKey) throws IOException {
    lock.obtainReadLock();
    try {
      RecordSearcher recordSearcher = new RecordSearcher(this);
      return recordSearcher.search(rowKey, columnName, cellKey);
    } finally {
      lock.releaseReadLock();
    }
  }

  public RowColumnValues[] get(RowFilter filter) throws IOException {
    lock.obtainReadLock();
    try { 
      RecordSearcher recordSearcher = new RecordSearcher(this);
      return recordSearcher.search(filter);
    } finally {
      lock.releaseReadLock();
    }
  }
  
  /**
   * 데이터를 조회한다.
   * !!!주의사항!!!!
   * return되는 ColumnValue 값의 rowKey에는 값이 없다. 
   * 이것은 실시간 처리의 경우 데이터 전송량이 많아지는 것을 방지하기 위해서이다. 
   * @param rowKey
   * @param columnName
   * @return
   * @throws IOException
   */
  public ColumnValue[] get(Row.Key rowKey, String columnName) throws IOException {
    lock.obtainReadLock();
    try { 
      RecordSearcher recordSearcher = new RecordSearcher(this);
      return recordSearcher.search(rowKey, columnName);
    } finally {
      lock.releaseReadLock();
    }
  }
  
  /**
   * 특정 Row의 지정된 컬럼들의 데이터를 조회한다.
   * row에 저장된 컬럼의 데이터가 정해진 사이즈보다 많은 경우 Exception이 발생한다.
   * @param rowKey
   * @param columnNames
   * @return
   * @throws IOException
   */
  public ColumnValue[][] get(Row.Key rowKey, String[] columnNames) throws IOException {
    lock.obtainReadLock();
    try { 
      RecordSearcher recordSearcher = new RecordSearcher(this);
      return recordSearcher.search(rowKey, columnNames);
    } finally {
      lock.releaseReadLock();
    }
  }
  
  /**
   * 특정 Row의 모든 컬럼의 데이터를 조회한다.
   * @param rowKey
   * @return
   * @throws IOException
   */
  public ColumnValue[][] get(Row.Key rowKey) throws IOException {
    //FIXME StopAction이 실행중인 경우에는 Exception을 보낸다.
    //(StopAction중인 경우에는 다른 서버로 Tablet이 이관중이고 이관받은 서버가 MetaData를 수정했음에도 불구하고
    // client에 있는 cache에 의해 계속 이전서버로 데이터를 요청할 수 있기 때문이다.
    //Client에서는 get 수행시 Exception이 발생하면 retry를 한다.
    
    //FIXME Column 순서에 대한 정의(사용자가 생성한 순서 or 알파벳순)
    int columnCount = table.getColumns().size();
    String[] columnNames = table.getColumns().toArray(new String[columnCount]);
    
    return get(rowKey, columnNames);
  }

  /**
   * 입력받은 rowKey와 가장 가까운 Row.Key중 가장 큰 Row.Key를 가지는 row의 값을 조회한다.
   * 주로 Root, Meta 테이블의 데이터를 이용하여 TabletInfo를 조회할 때 사용된다. 
   * @param rowKey
   * @return
   * @throws IOException
   */
  public ColumnValue getClosestMeta(Row.Key rowKey) throws IOException {
//    SortedSet<ColumnValue> closestRow.Keys = new TreeSet<ColumnValue>();
    
    //메모리로부터 삭제되지 않은 rowKey보다 크거나 같은 값 중 가장 작은 값을 가져 온다. 
    ColumnValue closestValueFromMemory = memorySSTable.findClosest(rowKey, META_COLUMN_NAME_TABLETINFO);
    ColumnValue closestValueFromDisk = diskSSTable.findClosestMeta(rowKey, META_COLUMN_NAME_TABLETINFO, true);
    
    while(true) {
      //Disk에서 가져온 값이 삭제된 값인 경우 다시 가져온다. 
      if(closestValueFromDisk != null && 
          memorySSTable.isDeleted(closestValueFromDisk.getRowKey(), META_COLUMN_NAME_TABLETINFO)) {
        closestValueFromDisk = diskSSTable.findClosestMeta(closestValueFromDisk.getRowKey(), META_COLUMN_NAME_TABLETINFO, false);
      } else {
        break;
      }
    }
    
    if(closestValueFromMemory == null && closestValueFromDisk == null) {
      LOG.debug("closestRowKey is null:" + rowKey);
      memorySSTable.print();
      return null;
    } else if(closestValueFromMemory == null && closestValueFromDisk != null) {
      return closestValueFromDisk;
    } else if(closestValueFromMemory != null && closestValueFromDisk == null) {
      return closestValueFromMemory;
    } else {
      if(closestValueFromMemory.getRowKey().compareTo(closestValueFromDisk.getRowKey()) <= 0) {
        return closestValueFromMemory;
      } else {
        return closestValueFromDisk;
      }
    }
  }
  
  /**
   * MemTablet의 현재 메모리 사이즈를 반환한다.
   * @return
   */
  public int getMemtableSize() {
    lock.obtainReadLock();
    
    try {
      return memorySSTable.getTabletSize();
    } finally {
      lock.releaseReadLock();
    }
  }

  /**
   * Row를 생성하거나 Row의 값을 변경하는 트렌젝션을 시작한다.timestamp의 값은 system 시간을 이용한다.
   * @param rowKey
   * @return
   */
  public TxId startTx(Row.Key rowKey) throws IOException {
    return startTx(rowKey, true);
  }

  /**
   * 
   * @param rowKey
   * @param systemTimestamp
   * @return
   */
  public TxId startTx(Row.Key rowKey, boolean systemTimestamp) throws IOException {
    //LOG.debug("startTx rowKey:" + rowKey);
    if(!ready) {
      //LOG.debug("Tablet Not Ready. StartTx return null TxId");
      throw new IOException("Tablet is not ready");
    }
    
    if(rowKey.compareTo(endRowKey) > 0) {
      throw new IOException(rowKey + " is out of range:" + tabletInfo);
    }
    
    lock.obtainWriteLock();
    try {
      //startTx의 경우 lock이 걸려 있는 경우 client로 바로 return 처리한다.
      //NeptuneClient에서 다시 retry 하는 방식으로 변경
      //이유는 client에게 무한정 대기하라고 할 수 없기 때문이다.
      if(lockedRow.containsKey(rowKey)) {
        LOG.info(" rowkey[" + rowKey + "] in tablet[" + tabletInfo.getTabletName() + "] is locked. RETRY");
        return null;
      }

      TxId txId = TxId.generate(tabletInfo.getTabletName(), rowKey, systemTimestamp);

      TxAction txAction = new TxAction(txId);
      if(!actionChecker.startAction(txAction)) {
        String actionNameList = "";
        for(String actionName : actionChecker.getAllActions()) {
          actionNameList += (actionName + ", ");
        }
        
        LOG.info("TxAction to rowkey[" + rowKey + "] in tablet[" + tabletInfo.getTabletName() + "] is blocked");
        LOG.info("ActionList :" + actionNameList);
        return null;
      }
    
      txMap.put(txId, new TxValue(rowKey));
      lockedRow.put(rowKey, new Long(System.currentTimeMillis()));
      return txId;
    } finally {
      lock.releaseWriteLock();
    }
  }

  public void removeRow(TxId txId, long timestamp) throws IOException {
    //FIXME Thread로 처리할 것인지 그냥 처리할 것인지..
    //그냥 처리할 경우 Row의 column 값이 많은 경우 client에서 timeout exception이 발생할 수 있음.
    
    TxValue txValue = getTxValue(txId);
    Row.Key rowKey = txValue.getRowKey();
    
    if(timestamp == 0) {
      timestamp = System.currentTimeMillis();
    }
    
//    boolean isdeleted = false;
    for(String columnName: table.getColumns()) {
      //FIXME 값을 제대로 조회하지 못하는 경우가 발생. 특히 META 삭제 처리를 할 경우
      ColumnValue[] columnValues = get(rowKey, columnName);
      
      //LOG.info("start RemoveRow:" + tabletInfo.getTabletName() + "," + columnName + "," + columnValues);
      if(columnValues == null || columnValues.length == 0) {
        continue;
      }
    
      for(ColumnValue columnValue : columnValues) {
        //get() 메소드의 경우 columnValue의 rowKey에는 값이 없다.
        //LOG.info("removeRow:" + columnValue);
        columnValue.setRowKey(rowKey);
        deleteColumnValue(txId, columnName, columnValue.getCellKey(), timestamp);
//        isdeleted = true;
      }
    }
    
//    if(!isdeleted && "META".equals(tabletInfo.getTableName())) {
//      LOG.info("Can't delete META :" + rowKey);
//    }
  }
  
  /**
   * 특정 row에 컬럼 데이터를 저장한다.
   * @param txId
   * @param columnName
   * @param columnValues
   */
  public void insertColumnValue(final TxId txId, final String columnName, final ColumnValue[] columnValues) throws IOException {
    TxValue txValue = getTxValue(txId);
    if(!table.getColumns().contains(columnName)) {
      throw new TxException("Invalid column [" + columnName + "]");
    }
    
    for(ColumnValue record: columnValues) {
      long timestamp = txId.isSystemTimestamp() ? System.currentTimeMillis() : record.getTimestamp();
      if(timestamp < 0) {
        continue;
      }      
      
      ChangeLog changeLog = new ChangeLog(Constants.LOG_OP_ADD_COLUMN_VALUE, 
                        tabletInfo.getTableName(), 
                        txValue.getRowKey(),
                        columnName,
                        record.getCellKey(),
                        timestamp,
                        record.getValue());     
      txValue.addChangeLog(changeLog);
    }   
  }

  /**
   * 특정 Row의 컬럼 값을 삭제한다.
   * @param txId
   * @param columnName
   * @param columnKey
   * @param timestamp
   */
  public boolean deleteColumnValue(TxId txId, String columnName, Cell.Key columnKey, long timestamp) throws IOException {
    TxValue txValue = getTxValue(txId);
    if(!table.getColumns().contains(columnName)) {
      throw new TxException("Invalid column [" + columnName + "]");
    }
//    ColumnValue columnValue = this.get(txValue.getRowKey(), columnName, columnKey);
//    if(columnValue == null) {
//      return false;
//    }
    ChangeLog changeLog = new ChangeLog(Constants.LOG_OP_DELETE_COLUMN_VALUE, 
                      tabletInfo.getTableName(), 
                      txValue.getRowKey(),
                      columnName,
                      columnKey,
                      timestamp,
                      null);      
    txValue.addChangeLog(changeLog);
    
    return true;
  }

  private TxValue getTxValue(TxId txId) throws IOException {
    TxValue txValue = null;
    lock.obtainReadLock();
    try {
      txValue = txMap.get(txId);
    } finally {
      lock.releaseReadLock();
    }
    
    if(txValue == null) {
      throw new TxException("Tx [" + txId + "] not started");
    }
    return txValue;
  }
  
  /**
   * 트렌젝션을 취소시킨다. 트렌젝션 시작 후 입력된 값은 반영되지 않는다.
   * @param txId
   */
  public Row.Key rollback(TxId txId) throws IOException {
    TxValue txValue = getTxValue(txId);
    Row.Key rowKey = txValue.getRowKey();
    removeLockedRow(rowKey, txId);
    
    actionChecker.endAction(new TxAction(txId));
    
    return rowKey;
  }
  
  public void commit(TxId txId) throws IOException {
    commit(txId, true);
  }
  
  public void commit(TxId txId, boolean saveLog) throws IOException {
    TxValue txValue = getTxValue(txId);
	
    internalCommit(txId, txValue.getRowKey(), txValue.getChangeLogs().toArray(new ChangeLog[0]), saveLog);
  }
  
  /**
   * 트렌젝션을 종료한다.
   * @param txId
   * @throws IOException
   */
  public void commit(TxId txId, ChangeLog[] changeLogList, boolean saveLog) throws IOException {
	internalCommit(txId, getTxValue(txId).getRowKey(), changeLogList, saveLog);
  }
  
  public boolean apply(Row.Key rowKey, TxId txId, ChangeLog[] changeLogList, boolean saveLog) throws IOException {
    if(!ready || droped) {
      throw new IOException("Tablet is not ready:ready=" + ready + ", droped=" + droped);
    }
    
    if(rowKey.compareTo(endRowKey) > 0) {
      throw new IOException(rowKey + " is out of range:" + tabletInfo);
    }
    
    ApplyAction applyAction = null;
    lock.obtainWriteLock();
    try {
      if(lockedRow.containsKey(rowKey)) {        
        LOG.info("rowKey : " + rowKey + " is locked. " +
        		"Try applying in next time.lockedRow.size=" + lockedRow.size() + ",txMap.size=" + txMap.size() + "," + tabletInfo.getTabletName());
		  // by sangchul
		  // false를 반환할 경우 Client가 다시 시도
        return false;
      }
    
      applyAction = new ApplyAction(txId);
      
      if(!actionChecker.startAction(applyAction)) {
        LOG.info("Cannot start apply action. Try applying in next time:" + tabletInfo.getTabletName() + "," + rowKey);
          // by sangchul
          // false를 반환할 경우 Client가 다시 시도
        return false;
      }
  
      txMap.put(txId, new TxValue(rowKey));
      lockedRow.put(rowKey, new Long(System.currentTimeMillis()));
    } finally {
      lock.releaseWriteLock();
    }

    try {
      return mutateMemorySSTable(txId, rowKey, changeLogList, saveLog);
    } finally {
      removeLockedRow(rowKey, txId);
      actionChecker.endAction(applyAction);
    }
  }

  private void removeLockedRow(Row.Key rowKey, TxId txId) {
    lock.obtainWriteLock();
    try {
      txMap.remove(txId);
      lockedRow.remove(rowKey);
    } finally {
      lock.releaseWriteLock();
    }
  }

  private void internalCommit(TxId txId, Row.Key rowKey, ChangeLog[] changeLogList, boolean saveLog) throws IOException {
    CommitAction commitAction = new CommitAction(txId);
    if(!actionChecker.startAction(commitAction)) {
      LOG.error("fail commit cause actionChecker");
      throw new IOException("fail commit");
    }

    try {
      mutateMemorySSTable(txId, rowKey, changeLogList, saveLog);
    } finally {
      actionChecker.endAction(commitAction);      
    }
    
    removeLockedRow(rowKey, txId);
        
    actionChecker.endAction(commitAction);
    actionChecker.endAction(new TxAction(txId));
  }
  
  public void setInChangeLogError(boolean flag) {
    inChangeLogError.set(flag);
  }
  
  private boolean mutateMemorySSTable(TxId txId, Row.Key rowKey, ChangeLog[] changeLogList, boolean saveLog) throws TxException {
    try {
      if (inChangeLogError.get() == false) {
        memorySSTable.commit(txId, changeLogList, saveLog);
      } else {
        //LOG.warn("Can't commit cause by ChangeLog Error:" + tabletInfo.getTabletName() + "," + rowKey);
      }
      // by sangchul
      // change log 에러로 minor compaction이 수행중이면 
      // 클라이언트가 잠시 후 다시 시도하도록 한다.
      return !inChangeLogError.get(); 
    } catch (ChangeLogInterruptedException e) {
      Thread.interrupted();
      LOG.info("[" + Thread.currentThread().getName() + "] Thread is interrupted in mutating. Retry by client");
      return false;
    } catch (ChangeLogException e) {
      if (inChangeLogError.getAndSet(true) == false) {
        //ChangeLog 에러가 발생하면 minor compaction을 수행시킨다.
        LOG.error("Exception in mutateMemorySSTable tabletName [" 
            + this.tabletInfo.getTabletName() + "]" + e);
        forceToCancelAllTxes();
        forceToRunMinorCompaction();
      }
      
      //호출한 client로는 TxException을 보낸다.
      throw new TxException(e.getMessage());
    } catch (Exception e) {
      LOG.error("Error while mutateMemorySSTable:" + tabletInfo, e);
      throw new TxException(tabletServer.getHostName() + ":" + e.getMessage());
    }
  }
  
  private void forceToCancelAllTxes() {
    Collection<TxValue> txValueList = txMap.values();
    LOG.info("force to cancel all transactions txValueList size : " + txValueList.size());
    
    lock.obtainWriteLock();
    for(TxValue tx : txValueList) {
      LOG.info("interrupt thread [" + tx.txOwnerThread + "]");
      tx.txOwnerThread.interrupt();
    }
    lock.releaseWriteLock();
  }

  private void forceToRunMinorCompaction() {
    LOG.info("force to run minor compaction");
    
    int retryCount = 0;
    while(retryCount++ < 100) {
      if(actionChecker.startAction(minorCompactionAction)) {
        break;
      }
      
      try {
        Thread.sleep(2000);
      } catch (InterruptedException e1) {
        return;
      }
    }
    
    if (retryCount >= 100) {
      List<String> actionList = actionChecker.getAllActions();
      
      for(String actionInfo : actionList) {
        LOG.fatal(actionInfo);          
      }
      
      LOG.fatal("Kill TabletServer due to failure of starting minorCompaction");
      System.exit(-1);
    }
  }
  
  /**
   * META/ROOT tablet에 split된 tablet 정보를 저장한다.
   * 3개의 작업(1 delete, 2 insert)이 하나의 change log에 저장되도록한다.
   * @param txId
   * @param splitedTablets
   * @throws IOException
   */
  public void saveTabletSplitedInfo(TxId txId, TabletInfo targetTablet, TabletInfo[] splitedTablets) throws IOException {
    TxValue txValue = getTxValue(txId);

    ByteArrayOutputStream bout = new ByteArrayOutputStream();
    DataOutput out = new DataOutputStream(bout);
    targetTablet.write(out);
    for(int i = 0; i < splitedTablets.length; i++) {
      splitedTablets[i].write(out);
    }
    ChangeLog changeLog = new ChangeLog(Constants.LOG_OP_MODIFY_META, 
                        tabletInfo.getTableName(), 
                        txValue.getRowKey(),
                        META_COLUMN_NAME_TABLETINFO,
                        new Cell.Key(""),
                        System.currentTimeMillis(),
                        bout.toByteArray());     
    txValue.addChangeLog(changeLog);
  }  
  
  /**
   * MemoryTablet이 일정 사이즈보다 커지게 되면 Memory의 내용을 Disk의 MapFile로 저장한다.
   * @throws IOException
   */
  public void doMinorCompaction() throws IOException {
    if(this.getMemtableSize() == 0) {
      return;
    }
    //LOG.debug("MinorCompaction Target:" + tabletInfo.getTabletName());
    if(!actionChecker.startAction(minorCompactionAction)) {
//      LOG.debug("Can't run minor compaction:" + tabletInfo.getTabletName()); 
//      for(String eachAction: actionChecker.getRunningActions()) {
//        if(eachAction.indexOf("ApplyAction") >= 0) {
//          continue;
//        }
//        LOG.debug(tabletInfo.getTabletName() + " run action:" + eachAction);
//      }
//      for(String eachAction: actionChecker.getWaitingActions()) {
//        if(eachAction.indexOf("ApplyAction") >= 0) {
//          continue;
//        }
//        LOG.debug(tabletInfo.getTabletName() + " wait action:" + eachAction);
//      }
      
      return;
    }
  }
  
  public boolean neededMajorCompaction() {
	  return getMapFileCount() > 2;
  }
   
  public boolean isDroped() {
    return this.droped;
  }

  private void clearAll() throws IOException {
    for(Map.Entry<String, TabletScanner> entry: tabletScanners.entrySet()) {
      try {
        ScannerAction action = new ScannerAction(entry.getKey());
        actionChecker.endAction(action);
        entry.getValue().close();
      } catch (IOException e) {
        LOG.error("Can't close scanner while drop:" + tabletInfo, e);
      }
    }
    
    actionChecker.endActions();
    //TODO Tablet이 compaction 중인 경우에는 어떻게 처리해야 하나? -> drop이 호출되기전에 모두 종료된 상태임

    tabletScanners.clear();

    droped = true;
    ready = false;
    
    //memory 정보 삭제
    memorySSTable.clearAllMemory();
    diskSSTable.clearAllMemory();
    
    lockedRow.clear();
    txMap.clear();
  }
  
  public void drop() throws IOException {
    clearAll();
    
    //change log 삭제
    deleteChangeLog();    
  }  
  
  /**
   * 하나의 Tablet에서 관리하고 있는 여러개의 MapFile을 merge하여 하나의 MapFile로 만든다.
   * @throws IOException
   */
  public void doMajorCompaction() throws IOException {
    if(!actionChecker.startAction(majorCompactionAction)) {
      return;
    }
  }

  public void doTabletSplit() throws IOException {
//    lock.obtainReadLock();
//    try {
//      //if(!neededSplit()) {
//      if(getMapFileCount() != 1) {
//        return;
//      }
//    } finally {
//      lock.releaseReadLock();
//    }
    
    actionChecker.startAction(tabletSplitAction);
  }
  
  public boolean neededSplit() throws IOException {
    return neededSplit(0.9);
  }
  
  public boolean neededSplit(double checkRate) throws IOException {
    lock.obtainReadLock();
    try {
      if(majorCompactionAction != null && !majorCompactionAction.isEnd()) {
        return false;
      }
      //Max Disk 크기의 80%가 되면 split 수행
      //Memory의 경우 실제 Class 오버헤더가 있기 때문에 30%만 적용
      return (sumMapFileSize() + getMemtableSize() * 0.3) >= (maxTabletSizeByte * checkRate);
    } catch (IOException e) {
      LOG.warn("Error checking neededSplit:" + e.getMessage());
      return false;
    } finally {
      lock.releaseReadLock();
    }
  }
  
  public int getMapFileCount() {
//    lock.obtainReadLock();
//    try {
//      return diskSSTable.getMaxMapFileCount();
//    } finally {
//      lock.releaseReadLock();
//    }
    return diskSSTable.getMaxMapFileCount();
  }
  
  public long sumMapFileSize() throws IOException {
    lock.obtainReadLock();
    try {
      return diskSSTable.sumMapFileSize();
    } finally {
      lock.releaseReadLock();
    }
  }
  
  /**
   * Client API에서는 ColumnRecord 하나만 반환하도록 하지만 네트워크 연결 횟수를 줄이기 위해
   * 한번에 fetch하는 사이즈 만큼 전송한다.
   * @param scannerId
   * @return
   * @throws IOException
   */
  public ColumnValue[] scanNext(String scannerId) throws IOException {
    //lock.obtainReadLock();
    TabletScanner tabletScanner = null;
    try {
      tabletScanner = tabletScanners.get(scannerId);
      if (tabletScanner == null) {
        throw new IOException("[" + scannerId + "] not opened");
      }
    } finally {
      //lock.releaseReadLock();
    }
    return tabletScanner.next();
  }

  /**
   * get의 경우 호출될 때마다 매번 MapFile을 open한 후 seek 해야 하기 때문에 
   * 대량의 데이터를 sequential하게 read하는데에는 부적합하다.
   * Scanner의 경우 scanner를 open할 때 MapFile을 열어 seek 한 후 next() 호출에 의해 파일 포인터를 이동시키면서
   * 데이터를 조회한다.
   * 
   *  Scanner는 Minor, Major compaction, tablet split, merge 중인 경우에는 open할 수 없다. 이 경우 바로 null을 반환하고
   *  클라이언트 모듈(NeptuneClient)에서 여러번 retry하도록 한다. 이유는 Split 되는 경우 Spilit 종료 후에는 해당 Row Range에 대한
   *  Tablet Server가 변경되어 있기 때문이다.
   * @param startRowKey
   * @param endRowKey
   * @param columnNames
   * @return
   * @throws IOException
   */
  public TabletScanner openScanner(String columnName, RowFilter scanFilter) throws IOException {
    String scannerId = TabletScanner.generateScannerId(tabletInfo);
    
    ScannerAction action = new ScannerAction(scannerId);

    //Minor, Major Compaction 중인 경우 open하지 않고 return
    if(!actionChecker.startAction(action))  {
      LOG.info(tabletInfo.getTabletName() + ": Can't open scanner while TabletStop/Compaction/Split running");
      return null;
    }
    
    try {
      CellFilter cellFilter = null;
      if(scanFilter.getCellFilters().size() > 0) {
        for(CellFilter eachCellFilter: scanFilter.getCellFilters()) {
          if(columnName.equals(eachCellFilter.getColumnName())) {
            cellFilter = eachCellFilter;
            break;
          }
        }
      }
      
      if(cellFilter == null) {
        cellFilter = new CellFilter(columnName);
      }
      
      TabletScanner tabletScanner = new TabletScanner( scannerId,
                                                        conf,
                                                        tabletInfo, 
                                                        scanFilter.getStartRowKey(), 
                                                        scanFilter.getEndRowKey(), 
                                                        columnName,
                                                        memorySSTable, diskSSTable,
                                                        cellFilter,
                                                        table.getNumOfVersion());
      
      tabletScanners.put(tabletScanner.getScannerId(), tabletScanner);
      return tabletScanner;
    } catch (IOException e) {
      actionChecker.endAction(action);
      throw e;
    }
  }

  /**
   * open되어 있는 scanner를 close한다.
   * @param scannerId
   * @throws IOException
   */
  public void closeScanner(String scannerId) throws IOException {
    ScannerAction action = new ScannerAction(scannerId);
    
    //lock.obtainWriteLock();
    
    try { 
      TabletScanner tabletScanner = tabletScanners.get(scannerId);
      if (tabletScanner == null) {
        actionChecker.endAction(action);
        throw new IOException("[" + scannerId + "] not opened");
      }
      tabletScanner.close();
      tabletScanners.remove(scannerId);
      //tabletScanners.notifyAll();
    } finally {
      //lock.releaseWriteLock();
    }
    
    actionChecker.endAction(action);
  }
  
  public TabletInfo getTabletInfo() {
    return tabletInfo;
  } 
  
  /**
   * 
   * @return
   */
  public int getOpenedScannerCount() {
    //lock.obtainReadLock();
    try {
      return tabletScanners.size();
    } finally {
      //lock.releaseReadLock();
    }
  }
  
  /**
   * 
   * @return
   */
  public TabletReport getTabletDetailInfo() throws IOException {
    //LOG.debug("Start getTabletDetailInfo:" + tabletInfo);
    TabletReport tabletReport = new TabletReport();
    tabletReport.setTabletInfo(tabletInfo);
    
    tabletReport.setRunningActions(actionChecker.getRunningActions().toArray(new String[]{}));
    tabletReport.setWaitingActions(actionChecker.getWaitingActions().toArray(new String[]{}));
    
    tabletReport.setMemoryTabletSize(memorySSTable.getTabletSize());
    tabletReport.setMemoryTabletDataCount(memorySSTable.getDataCount());
    
    tabletReport.setColumnMapFileCount(diskSSTable.getMapFileCount());
    tabletReport.setColumnMapFileSize(diskSSTable.getMapFileSize());
    tabletReport.setColumnIndexSize(diskSSTable.getMapFileIndexSize());
    
    tabletReport.setServerMaxMemory(Runtime.getRuntime().maxMemory());
    tabletReport.setServerFreeMemory(Runtime.getRuntime().freeMemory());
    tabletReport.setServerTotalMemory(Runtime.getRuntime().totalMemory());
    
    //System.out.println(Runtime.getRuntime().maxMemory() + "," + Runtime.getRuntime().freeMemory() + "," + Runtime.getRuntime().totalMemory());
    return tabletReport;
  }

  /**
   * 
   * @throws IOException
   */
  public void deleteTablet() throws IOException {
    //LOG.debug("Drop:" + tabletInfo.getTabletName());
    lock.obtainWriteLock();
    try {

      clearAll();
      
      try {
        //delete tablet data
        GPath dataPath = Tablet.getTabletPath(conf, tabletInfo);
        long startTime = System.currentTimeMillis();
        while(true) {
          boolean result = fs.delete(dataPath, true);
          if(result) {
            break;
          }
          try {
            Thread.sleep(100);
          } catch (InterruptedException e) {
            return;
          }
          if(System.currentTimeMillis() - startTime > 30 * 1000) {
            LOG.warn("Can't delete " + dataPath + " while droping " + tabletInfo);
            break;
          }
        }
      } catch (IOException e) {
        LOG.error("found exception while deleting data file file for drop", e);
        e.printStackTrace(System.out);
        throw e;
      }
      
      //delete tablet change log
      try {
        memorySSTable.getChangeLogFileSystem().delete(tabletInfo.getTabletName());
      } catch (IOException e) {
        LOG.error("found exception while deleting change log file for drop", e);
        throw e;
      }
    } finally {
      lock.releaseWriteLock();
    }
  }

  public void deleteChangeLog() throws IOException {
    deleteChangeLog(this.memorySSTable.getChangeLogFileSystem(),
        tabletInfo.getTableName(),
        tabletInfo.getTabletName());
  }
  
  public static void deleteChangeLog(ChangeLogFileSystemIF changeLogFileSystem, 
                                        String tableName, 
                                        String tabletName) throws IOException {
    //TODO ChangeLog도 삭제하지 않고 Trash로 이동
    try {
      if(changeLogFileSystem.exists(tabletName)) {
        changeLogFileSystem.delete(tabletName);
      }
    } catch (IOException e) {
      LOG.error("Can't delete ChangeLog file while drop:" + e.getMessage(), e);
      throw e;
    }
  }
  
  public void setReady(boolean ready) {
    lock.obtainWriteLock();
    try {
      this.ready = ready;
    } finally {
      lock.releaseWriteLock();
    }
  }
  /**
   * 
   * @return
   */
  public boolean isReady() {
    lock.obtainReadLock();
    try {
      return ready;
    } finally {
      lock.releaseReadLock();
    }
  }
  
  public void deleteSplitTempDir() throws IOException {
    fs.delete(Tablet.getTabletSplitTempPath(conf, tabletInfo), true);
  }

//  public boolean readForStop(ReadyForStopListener readyForStopListener) throws IOException {
//    TabletStopAction action = new TabletStopAction(tabletInfo, readyForStopListener);
//    return actionChecker.startAction(action);
//  }
//
//  public void cancelReadyForStop() throws IOException {
//    //LOG.debug("cancelReadyForStop:" + tabletInfo.getTabletName());
//    TabletStopAction action = new TabletStopAction(tabletInfo, null);
//    actionChecker.endAction(action);
//  }
  
  public void truncateColumn(String columnName) throws IOException {
    if(memorySSTable.getTabletSize() > 0) {
      doMinorCompaction();
      while(true) {
        try {
          Thread.sleep(1*1000);
        } catch (InterruptedException e) {
        }
        if(minorCompactionAction.isEnd()) {
          break;
        }
      }
    }
    diskSSTable.truncateColumn(columnName);
  }
  
  public void addColumn(String addedColumnName) throws IOException {
    synchronized(table) {
      if(!table.getColumns().contains(addedColumnName)) {
        table.addColumn(addedColumnName);
      }
    }
  }
  
  public DiskSSTableIF getDiskSSTable() {
    return this.diskSSTable;
  }
  
  public MemorySSTableIF getMemorySSTable() {
    return this.memorySSTable;
  }
  
  public ActionChecker getActionChecker() {
    return this.actionChecker;
  }
  
  public String[] getAllActions() {
    List<String>  actions = actionChecker.getAllActions();
    return actions.toArray(new String[]{});
  }
  

  public boolean isEmpty() {
    return memorySSTable.isEmpty() && diskSSTable.isEmpty();
  }
  
  public boolean hasValue(String columnName, Row.Key rowKey) throws IOException {
    RecordSearcher recordSearcher = new RecordSearcher(this);
    return recordSearcher.hasValue(columnName, rowKey);
  }
  
  public static Row.Key generateMetaRowKey(String tableName, Row.Key rowKey) {
    if(tableName == null) {
      return Row.Key.MAX_KEY;
    }
    
    byte[] tableNameBytes = tableName.getBytes();
    byte[] metaRowKeyBytes = new byte[tableNameBytes.length + rowKey.getLength() + 1];
    System.arraycopy(tableNameBytes, 0, metaRowKeyBytes, 0, tableNameBytes.length);
    metaRowKeyBytes[tableNameBytes.length] = (byte)'.';
    System.arraycopy(rowKey.getBytes(), 0, metaRowKeyBytes, tableNameBytes.length + 1, rowKey.getLength());
    
    return new Row.Key(metaRowKeyBytes); 
  }
  
//  public static String getTableNameFromMetaRow.Key(Row.Key metaTabletRow.Key) {
//    String key = metaTabletRow.Key.getKey();
//    int index = key.indexOf(".");
//    if(index <= 0)  return null;  
//    return key.substring(0, index);
//  } 
  
//  public static GPath getRootTabletServerPath(NeptuneConfiguration conf) {
//    return new GPath(conf.get("neptune.root") + "/servers/root/");
//  }
  
  public static TabletInfo getRootTabletInfo(NConfiguration conf, LockService lockService) throws IOException {
    LockMetaData lockMetaData = lockService.getMetaData(Constants.ROOT_TABLET);
    if(lockMetaData == null) {
      LOG.warn("Lock meta data of pleiades[" + Constants.ROOT_TABLET + "] is null" 
          + ", so ROOT tablet info is also null");
      return null;
    }
    String contentKey = lockMetaData.getContentKey();
    if(contentKey == null) {
      return null;
    }
    
//    GPath rootTabletPath = new GPath(conf.get("neptune.root") + Constants.ROOT_TABLET );
//    
//    NeptuneFileSystem fs = NeptuneFileSystem.get(conf);
//    GPath[] paths = fs.list(rootTabletPath);
//    if(paths == null || paths.length == 0) {
//      return null;
//    }
//
//    String rootTabletLockMeta = paths[0].getName();
    
    int index = contentKey.indexOf(",");
    //처음 생성되어 TabletServer가 할당되지 않은 경우
    String tabletName;
    String rootTabletHost;
    if(index > 0) {
      tabletName = contentKey.substring(0, index);
      rootTabletHost = NetworkUtil.fileNameToHostName(contentKey.substring(index + 1));
    } else {
      tabletName = contentKey;
      rootTabletHost = null;
    }
    
    TabletInfo rootTabletInfo = new TabletInfo(TABLE_NAME_ROOT, tabletName, 
                                                Row.Key.MIN_KEY, Row.Key.MAX_KEY);
    
    rootTabletInfo.setAssignedHostName(rootTabletHost);
    
    return rootTabletInfo;
  }
  
  public static GPath getTabletPath(NConfiguration conf, TabletInfo tabletInfo) {
    return new GPath(TableSchema.getTableDataPath(conf, tabletInfo.getTableName()) + "/" + tabletInfo.getTabletName());
  }
  
  public static GPath getTabletPath(NConfiguration conf, String tableName, String tabletName) {
    return new GPath(TableSchema.getTableDataPath(conf, tableName) + "/" + tabletName);
  }
  
  public static String generateTabletName(String tableName) {
    return tableName + "_" + IdGenerator.getId();
  }
  
  public static GPath getTabletSplitTempPath(NConfiguration conf, String tableName, String tabletName) {
    return new GPath(getTabletPath(conf, tableName, tabletName) + DIR_TEMP_SPLIT);
  }
  
  public static GPath getTabletSplitTempPath(NConfiguration conf, TabletInfo tabletInfo) {
    return new GPath(getTabletPath(conf, tabletInfo) + DIR_TEMP_SPLIT);
  }

  public static GPath getTabletMajorCompactionTempPath(NConfiguration conf, String tableName, String tabletName) {
    return new GPath(getTabletPath(conf, tableName, tabletName) + DIR_TEMP_MAJOR);
  }

  public static GPath getTabletMajorCompactionTempPath(NConfiguration conf, TabletInfo tabletInfo) {
    return new GPath(getTabletPath(conf, tabletInfo) + DIR_TEMP_MAJOR);
  }

  public static GPath getTabletMinorCompactionTempPath(NConfiguration conf, String tableName, String tabletName) {
    return new GPath(getTabletPath(conf, tableName, tabletName) + DIR_TEMP_MINOR);
  }

  public static GPath getTabletMinorCompactionTempPath(NConfiguration conf, TabletInfo tabletInfo) {
    return new GPath(getTabletPath(conf, tabletInfo) + DIR_TEMP_MINOR);
  }
  
  public static GPath getTabletLocalTempPath(NConfiguration conf, TabletInfo tabletInfo, String actionId) {
//    String tabletLocalTempPath = conf.get("neptune.local.temp") + "/" + 
//                                  System.currentTimeMillis() + "_" + 
//                                  tabletInfo.getTableName() + "_" +
//                                  tabletInfo.getTabletName();

    String tabletLocalTempPath = conf.get("neptune.local.temp") + "/" + actionId;

    return new GPath(tabletLocalTempPath + DIR_TEMP_UPLOAD);
  }
  
  public static GPath getTabletUploadTempPath(NConfiguration conf, TabletInfo tabletInfo, String actionId) {
    return new GPath(getTabletPath(conf, tabletInfo) + DIR_TEMP_UPLOAD + "/" + actionId);
  }

  public static String getTableNameFromTabletName(String tabletName) {
    //FIXME 나중에 Meta 테이블에 Table명 저장하는 형태로 변경
    int index = tabletName.lastIndexOf("_");
    if(index < 0) return null;
    return tabletName.substring(0, index);
  }
  
  public TabletServerIF getTabletServer() {
    return tabletServer;
  }

  public Row.Key[] getSplitedRowKeyRanges(int splitPerTablet) throws IOException {
    if(splitPerTablet < 0) {
      return null;
    }
    
    SortedSet<Row.Key> rowKeys = memorySSTable.getAllRowKeys();
    SortedSet<MapFileIndexRecord> indexRecords = diskSSTable.getMapFileIndex();
    
    for(MapFileIndexRecord eachIndex: indexRecords) {
      rowKeys.add(eachIndex.getRowKey());
    }
    
    int rowKeySize = rowKeys.size();
    
    if(splitPerTablet >= rowKeySize) {
      return rowKeys.toArray(new Row.Key[]{});
    } else {
      Row.Key[] result = new Row.Key[splitPerTablet];
      
      Row.Key[] rowKeyArr = new Row.Key[rowKeySize];
      rowKeys.toArray(rowKeyArr);
      int gap = rowKeySize / splitPerTablet;
      for(int i = 0 ; i < splitPerTablet - 1; i++) {
        result[i] = rowKeyArr[(i + 1) * gap];
      }
      result[splitPerTablet - 1] = tabletInfo.getEndRowKey();
      
      return result;
    }
  }
  
//  public static void main(String[] args) throws Exception {
//    testTabletServer(args);
//  }  
  
  static class TestGetThread extends Thread {
    NConfiguration conf = new NConfiguration();
    String threadId;
    Tablet tablet;
    Row.Key rowKey = new Row.Key("00000000010");
    TestGetThread(NConfiguration conf, String threadId, Tablet tablet) {
      this.conf = conf;
      this.threadId = threadId;
      this.tablet = tablet;
      
    }
    public void run() {
      try {
        int count = 0;
        long time = System.currentTimeMillis();
        while(true) {
          ColumnValue[][] columnValues = tablet.get(rowKey, new String[]{"Col1"});
          if(columnValues == null || columnValues.length == 0 || columnValues[0] == null) {
            System.out.println("Result is null");
            break;
          }
          count++;
          if(count % 1000 == 0) {
            System.out.println(threadId + ", Time:" + (System.currentTimeMillis() - time));
            time = System.currentTimeMillis();
          }
        }  
      } catch (Exception e) {
        e.printStackTrace();
      }
    }
  }
  
  static void testTabletServer(String[] args) throws Exception {
    int threadCount = 1;
    if(args.length >= 1) {
      threadCount = Integer.parseInt(args[0]);
    }
    NConfiguration conf = new NConfiguration();
    NTable ntable = NTable.openTable(conf, "M_INPUT_1000000");
    TabletInfo tabletInfo = ntable.getTabletInfo("M_INPUT_1000000_620585173539139");
    String hostName = tabletInfo.getAssignedHostName();
//    int index = hostName.indexOf(":");
//    String host = hostName.substring(0, index);
//    String port = conf.get("tabletServer.scanner.port");
//    hostName = host + ":" + port;
    System.out.println("HostName:" + hostName);
//    for(int i = 0 ;i < threadCount; i++) {
//      TestGetFromTabletServerThread t1 = new TestGetFromTabletServerThread(conf, "Thread" + (i + 1), null, hostName);
//      t1.start();
//      Thread.sleep(30 * 1000);
//    }
    
    DataServiceProtocol userTabletServer = (DataServiceProtocol) NRPC.getProxy(DataServiceProtocol.class,
        DataServiceProtocol.versionID, NetworkUtil.getAddress(hostName), conf);
    userTabletServer.test(threadCount, "AAA");
  }
  
  static class TestGetFromTabletServerThread extends Thread {
    NConfiguration conf = new NConfiguration();
    String threadId;
    Tablet tablet;
    Row.Key rowKey = new Row.Key("00000109308");
//    Row.Key rowKey = new Row.Key("00000000010");
    String hostName;
    TestGetFromTabletServerThread(NConfiguration conf, String threadId, Tablet tablet, String hostName) {
      this.conf = conf;
      this.threadId = threadId;
      this.tablet = tablet;
      this.hostName = hostName;
      
    }
    public void run() {
      try {
//        Socket socket = new Socket();
//        socket.connect(NetworkUtil.getAddress(hostName), 10 * 1000);
//        socket.setSoTimeout(10 * 1000);
//    
//        DataOutputStream out = new DataOutputStream(new BufferedOutputStream(socket.getOutputStream()));
//        out.write((byte) 61);
//        out.flush();
//    
//        DataInputStream in = new DataInputStream(new BufferedInputStream(socket.getInputStream()));
//        int count = 0;
//        long time = System.currentTimeMillis();
//        while(true) {
//          out.write((byte) 62);
//          out.flush();
//          ColumnValue columnValue = new ColumnValue();
//          columnValue.readFields(in);
//          count++;
//          if(count % 1000 == 0) {
//            System.out.println(threadId + ", Time:" + (System.currentTimeMillis() - time));
//            time = System.currentTimeMillis();
//          }
//        }
        
        DataServiceProtocol userTabletServer = (DataServiceProtocol) NRPC.getProxy(DataServiceProtocol.class,
                    DataServiceProtocol.versionID, NetworkUtil.getAddress(hostName), conf);
        int count = 0;
        long time = System.currentTimeMillis();
        while(true) {
          ColumnValue[][] columnValues = tablet.get(rowKey, new String[]{"Col1"});
          if(columnValues == null || columnValues.length == 0 || columnValues[0] == null) {
            System.out.println("Result is null");
            break;
          }
          count++;
          if(count % 1000 == 0) {
            System.out.println(threadId + ", Time:" + (System.currentTimeMillis() - time));
            time = System.currentTimeMillis();
          }
        }        
      } catch (Exception e) {
        e.printStackTrace();
      }
    }
  }

}
